package m3;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Scanner;

public class Producer {
    public static void main(String[] args) throws MQClientException {
        // 创建事务生产者实例
        TransactionMQProducer txMQproducer = new TransactionMQProducer("producerGroup3");
        // 设置name server
        txMQproducer.setNamesrvAddr("192.168.64.141:9876");
        // 设置事务消息监听器
        txMQproducer.setTransactionListener(new TransactionListener() {
            // 执行本地事务
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                System.out.println("执行本地事务，业务参数："+arg);
                Double r = Math.random();
                if (r < 0.3){
                    System.out.println("事务提交成功，向Rocketmq服务器提交");
                    return LocalTransactionState.COMMIT_MESSAGE;
                } else if (r < 0.6){
                    System.out.println("事务提交失败，向RocketMQ服务器撤回消息、回滚消息");
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                //执行本地事务的方法中一般不会出现未知状态
                //为了演示事务回查，提供UNKNOWN状态
                System.out.println("MQ对事务状态未知");
                return LocalTransactionState.UNKNOW;
            }
            // 处理 Rockermq 的事务状态回查
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.println("正在处理Rocketmq服务器的事务回查");
                return LocalTransactionState.UNKNOW;
            }
        });
        // 启动
        txMQproducer.start();
        // 发送事务消息 - 半消息 - 触发执行事务消息监听器来执行本地事务（自动触发）
        while (true){
            System.out.println("输入消息：");
            Message message = new Message("Topic3", "Tag3", new Scanner(System.in).nextLine().getBytes());

            //txMQproducer.sendMessageInTransaction(message,/*触发监听器，执行本地事务时，向它传递的业务数据参数*/args)
            txMQproducer.sendMessageInTransaction(message,"TestData");
        }
    }
}
